/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package aiaudio.processing.prediction.recalc;

import aiaudio.Application;
import aiaudio.database.tables.NeighbourGroupToUserTable;
import aiaudio.database.tables.NeighbourUserToGroupTable;
import java.io.IOException;
import java.util.Map;
import java.util.Map.Entry;
import java.util.NavigableMap;
import java.util.Vector;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;

/**
 *
 * @author nastya
 */
public class UserGroupAggregatedTableMapper extends TableMapper<ImmutableBytesWritable, UserGrouppingAggregate>{
    public static final String GROUP_TO_USER_MAP_NAME = "groupToUserMapName";

    HTable groupToUserMap;

    protected void processGroup(Entry<byte[], byte[]> entry, UserGrouppingAggregate aggregate) throws IOException {
        byte[] groupDistance = entry.getValue();
        byte[] groupNum = entry.getKey();            
        Vector<UserDistance> v = fillUSerList(groupNum);
        
        double dist = Bytes.toDouble(groupDistance);
        String gi = Bytes.toString(groupNum);
        aggregate.add(gi, dist, v);
    }

    protected Vector<UserDistance> fillUSerList(byte[] groupNum) throws IOException {
        Get g = new Get(groupNum);
        Result r = groupToUserMap.get(g);
        Vector<UserDistance>v = new Vector<UserDistance>();
        NavigableMap<byte[], byte[]> usersList = NeighbourGroupToUserTable.getUsers(r);
        for (Map.Entry<byte[], byte[]> e : usersList.entrySet()) {
            UserDistance ud = new UserDistance(Bytes.toString(e.getKey()), Bytes.toDouble(e.getValue()));
            v.add(ud);
        }
        return v;
    }

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        super.setup(context);
        String got = context.getConfiguration().get(GROUP_TO_USER_MAP_NAME);
        groupToUserMap = Application.database().getByName(got).getTable();
    }   
    
    
    @Override
    protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
        NavigableMap<byte[], byte[]> groupsList = NeighbourUserToGroupTable.getGroups(value);
        UserGrouppingAggregate aggregate = new UserGrouppingAggregate();

        for (Map.Entry<byte[], byte[]> entry : groupsList.entrySet()) {
            processGroup(entry, aggregate);
        }
     
        context.write(new ImmutableBytesWritable(key), aggregate);
    }
    
}
